/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.io;

import java.beans.IntrospectionException;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.InterruptedIOException;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.nio.charset.Charset;
import java.util.LinkedList;
import java.util.List;

import taskgraph.Config;
import taskgraph.ports.OutputPort;
import taskgraph.reflect.Accessor;
import taskgraph.reflect.BeanMetadata;
import taskgraph.tasks.PrimitiveTask;


/**
 * Source task for reading a delimited text file.
 *  
 * @author Armando Blancas
 * @see taskgraph.ports.OutputPort
 * @see java.lang.Class
 * @param <E> The type of elements read by this task.
 */
public class ReadDelimitedTextFile<E> extends PrimitiveTask {

	private static final char   DEFAULT_DELIMITER  = ',';
	private static final char   DEFAULT_QUALIFIER  = 0;
	private static final String DEFAULT_ENCODING   = Charset.defaultCharset().name();
	private static final int    DEFAULT_START_LINE = 1;
	private static final int    DEFAULT_CAPACITY   = Config.get().bufferCapacity();
	
	private OutputPort<E> output;
	private Class<E> clazz;
	private String filename;
	private String charsetName = DEFAULT_ENCODING;
	private char delimiter = DEFAULT_DELIMITER;
	private char qualifier = DEFAULT_QUALIFIER;
	private int startLine = DEFAULT_START_LINE;
	private int bufferSize = DEFAULT_CAPACITY;

	/**
	 * This constructor allows creating instances as beans.
	 */
    public ReadDelimitedTextFile() {
    	// do nothing
    }

    /**
     * ReadDelimitedTextFile constructor.
     * 
     * Creates a configured instance, with default encoding, 
     * delimiter, and qualifier characters.
     * 
     * @param output    The element output port.
     * @param clazz     The class literal of type parameter E.
     * @param filename  The full path of the file to read as String.
     */
    public ReadDelimitedTextFile(final OutputPort<E> output,
    	                         final Class<E> clazz,
    	                         final String filename) {
    	setOutput(output);
    	setClazz(clazz);
    	setFilename(filename);
    }
    
    /**
     * Creates a fully configured instance.
     * 
     * @param output       The element output port.
     * @param clazz        The class literal of type parameter E.
     * @param filename     The full path of the file to read as String.
     * @param charsetName  The name of the character encoding as String.
     * @param delimiter    The character that separates fields in the file.
     * @param qualifier    The character that surrounds fields in the file.
     * @throws UnsupportedEncodingException 
     */
    public ReadDelimitedTextFile(final OutputPort<E> output,
    	                         final Class<E> clazz,
    	                         final String filename,
    	                         final String charsetName,
    	                         final char delimiter,
    	                         final char qualifier)
    throws UnsupportedEncodingException {
    	setOutput(output);
    	setClazz(clazz);
    	setFilename(filename);
    	setCharsetName(charsetName);
    	setDelimiter(delimiter);
    	setQualifier(qualifier);
    }

	/**
	 * Gets the buffer size for reading chunks of characters.
	 * 
	 * @return the bufferSize
	 */
	public int getBufferSize() {
		return bufferSize;
	}

	/**
	 * Gets the name of the character encoding.
	 * 
	 * @return The charsetName value as {@code String}.
	 */
	public String getCharsetName() {
		return charsetName;
	}

	/**
	 * Gets the class object for the type parameter E.
	 * 
	 * @return The {@code Class<E>} value.
	 */
	public Class<E> getClazz() {
		return clazz;
	}

    /**
	 * Gets the character marker for field limits.
	 * 
	 * @return The delimiter {@code char} value.
	 */
	public char getDelimiter() {
		return delimiter;
	}
    
    /**
     * Gets the name of the file to read.
     * 
	 * @return The full path of the file as String
	 */
	public String getFilename() {
		return filename;
	}

	/**
     * Gets the output port.
     * 
	 * @return The element output port.
	 */
	public OutputPort<E> getOutput() {
		return output;
	}

	/**
	 * Gets the character that may surround field values.
	 * 
	 * @return The qualifier {@code char} value.
	 */
	public char getQualifier() {
		return qualifier;
	}

	/**
	 * Gets the line to start processing the file.
	 * 
	 * @return The first line to process, as an {@code int}
	 */
	public int getStartLine() {
		return startLine;
	}

	/**
	 * Parses a delimited text record.
	 * 
	 * @param line The file record as a {@code String}.
	 * @return A {@code String} array with field values in physical order.
	 */
	private String[] parseTextRecord(String line) {
		List<String> list = new LinkedList<String>();
		StringBuilder sb = new StringBuilder();
		boolean ignoreBlank = true;
		boolean qualified = false;
		for (char c : line.toCharArray()) {
			if (c == qualifier) {
				qualified = !qualified;
				ignoreBlank = !ignoreBlank;
			} else if (c == delimiter && !qualified) {
				list.add(sb.toString());
				sb = new StringBuilder();
				ignoreBlank = true;
			} else {
				if (c == ' ') {
					if (!ignoreBlank) sb.append(c);
				} else {
					sb.append(c);
					ignoreBlank = false;
				}
			}
		}
		list.add(sb.toString());
		return list.toArray(new String[0]);
	}
	
    /**
     * Performs the work of this task.
     * 
	 * Reads the input file, field by field, and creates an object E; 
	 * then writes the new object to the output port.
	 *<pre>
	 *open text file
	 *while not EOF
	 *    read a line from file
	 *    parse the fields in the line
	 *    create an instance of element E with the parsed fields
	 *    put the reference to E through the output port
	 *close text file
	 *close output port
	 *</pre>
	 */
	@Override
    public void run() {
		BufferedReader input = null;
    	try {
    		input = new BufferedReader(
    					new InputStreamReader(
    						new FileInputStream(filename), 
    						charsetName),
    					bufferSize);
    		
    		for (int i = startLine; i > 1; --i) {
    			String line = input.readLine();
    			if (line == null) {
    				break;
    			}
    		}
    		
    		Accessor[] setters = BeanMetadata.extractSetters(clazz);
    		while (!Thread.currentThread().isInterrupted()) {
    			String line = input.readLine();
    			if (line == null) {
    				break;
    			}
    			String[] values = parseTextRecord(line);
    			E element = BeanMetadata.newInstance(clazz, setters, values);
    			output.write(element);
            }
        } catch (InterruptedIOException e) {
        	// interruption request
        } catch (IOException e) {
            log.error("ReadDelimitedTextFile#run()", e);
        } catch (InstantiationException e) {
            log.error("ReadDelimitedTextFile#run()", e);
		} catch (IllegalAccessException e) {
            log.error("ReadDelimitedTextFile#run()", e);
		} catch (IntrospectionException e) {
            log.error("ReadDelimitedTextFile#run()", e);
		} catch (IllegalArgumentException e) {
            log.error("ReadDelimitedTextFile#run()", e);
		} catch (InvocationTargetException e) {
            log.error("ReadDelimitedTextFile#run()", e);
		} finally {
        	try {
        		if (input != null)
        			input.close();
                output.close();
			} catch (IOException e) {
				// best effort
			}
        }
    }

	/**
	 * Sets the buffer size for reading chunks of characters.
	 * <p>The argument should not be less than the default. If it is, the
	 * default size is used instead.
	 * 
	 * @param bufferSize the bufferSize to set
	 */
	public void setBufferSize(final int bufferSize) {
		this.bufferSize = (bufferSize < DEFAULT_CAPACITY) ? DEFAULT_CAPACITY : bufferSize;
	}

	/**
	 * Sets the name of the character encoding.
	 * 
	 * @param charsetName The {@code String} value to set as charset name.
	 * @throws UnsupportedEncodingException 
	 */
	public void setCharsetName(final String charsetName)
	throws UnsupportedEncodingException {
    	if (charsetName == null)
    		throw new IllegalArgumentException("charsetName == null");
		if (!Charset.isSupported(charsetName)) {
			throw new UnsupportedEncodingException (charsetName);
		}
		this.charsetName = charsetName;
	}

	/**
	 * Sets the class object for the type parameter E.
	 * 
	 * @param clazz The {@code Class<E>} value.
	 */
	public void setClazz(final Class<E> clazz) {
    	if (clazz == null)
    		throw new IllegalArgumentException("clazz == null");
		this.clazz = clazz;
	}

	/**
	 * Sets the character marker for field limits.
	 * 
	 * @param delimiter The {@code char} value that separates fields.
	 */
	public void setDelimiter(final char delimiter) {
		this.delimiter = delimiter;
	}

	/**
	 * Sets the name of the file to read.
	 * 
	 * @param filename The full path of the file as String.
	 */
	public void setFilename(final String filename) {
    	if (filename == null)
    		throw new IllegalArgumentException("filename == null");
		this.filename = filename;
	}

	/**
	 * Sets the output port.
	 * 
	 * @param output The element output port.
	 */
    public void setOutput(final OutputPort<E> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}

	/**
	 * Sets the character that may surround field values.
	 * 
	 * @param qualifier The qualifier {@code char} value.
	 */
	public void setQualifier(final char qualifier) {
		this.qualifier = qualifier;
	}

	/**
	 * Sets the line to start processing the file.
	 * <p>The default is 1 and processing starts at the
	 * first line in the file. If the input file has a header,
	 * set this property to 2, or as needed.
	 * 
	 * @param startLine the first line to process.
	 */
	public void setStartLine(int startLine) {
		this.startLine = startLine;
	}

}
